package cn.sciento.transfer.infra.rocketmq.service.impl;

import java.util.List;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.tools.command.SubCommandException;
import cn.sciento.core.base.BaseConstants;
import cn.sciento.transfer.app.service.ConsumerEventService;
import cn.sciento.transfer.app.service.EventErrorService;
import cn.sciento.transfer.app.service.TransferRedisService;
import cn.sciento.transfer.app.service.impl.ConsumerEventServiceImpl;
import cn.sciento.transfer.domain.entity.ConsDbConfig;
import cn.sciento.transfer.domain.entity.ConsTenantConfig;
import cn.sciento.transfer.domain.entity.ProducerConfig;
import cn.sciento.transfer.domain.repository.ConsDbConfigRepository;
import cn.sciento.transfer.domain.repository.ProducerConfigRepository;
import cn.sciento.transfer.infra.datasource.utils.TransferDataSourceTemplate;
import cn.sciento.transfer.infra.rocketmq.RocketmqConsumersGroupManager;
import cn.sciento.transfer.infra.rocketmq.config.MqConfigProperties;
import cn.sciento.transfer.infra.rocketmq.config.OffsetInitCapableRemoteStore;
import cn.sciento.transfer.infra.rocketmq.handler.MqMessageHandler;
import cn.sciento.transfer.infra.rocketmq.service.MqConsumerService;
import cn.sciento.transfer.infra.rocketmq.utils.RocketmqClientUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.stereotype.Service;

@Service
public class MqConsumerServiceImpl implements MqConsumerService {
  private Logger LOGGER = LoggerFactory.getLogger(MqConsumerServiceImpl.class);
  
  @Autowired
  private RocketmqConsumersGroupManager rocketmqConsumersGroupManager;
  
  @Autowired
  private TransferDataSourceTemplate transferDataSourceTemplate;
  
  @Autowired
  private MqConfigProperties mqConfigProperties;
  
  @Autowired
  private ProducerConfigRepository producerConfigRepository;
  
  @Autowired
  private ConsDbConfigRepository consDbConfigRepository;
  
  @Autowired
  private EventErrorService eventErrorService;
  
  @Autowired
  private AsyncTaskExecutor asyncTaskExecutor;
  
  @Autowired
  private TransferRedisService transferRedisService;
  
  public void startListener(String service, String tableName, ConsDbConfig consDbConfig, ConsTenantConfig consTenantConfig) {
    DefaultMQPushConsumer defaultMQPushConsumer = createConsumer(service, tableName, consDbConfig, consTenantConfig);
    OffsetInitCapableRemoteStore.notifyRemoteServerUpdateOffset(defaultMQPushConsumer.getNamesrvAddr(), defaultMQPushConsumer
        .getConsumerGroup(), 
        createTopicName(service, tableName), (consTenantConfig != null) ? consTenantConfig
        .getConsumerOffset() : consDbConfig.getConsumerOffset());
    MqMessageHandler mqMessageHandler = new MqMessageHandler(new ConsumerEventServiceImpl(this.eventErrorService), this.transferDataSourceTemplate, consDbConfig, consTenantConfig);
    this.rocketmqConsumersGroupManager.startConsumer(consDbConfig.generateConsumerGroup(consTenantConfig), mqMessageHandler);
  }
  
  public void startListenerWithOutUpdateOffset(String service, String tableName, ConsDbConfig consDbConfig, ConsTenantConfig consTenantConfig) {
    createConsumer(service, tableName, consDbConfig, consTenantConfig);
    MqMessageHandler mqMessageHandler = new MqMessageHandler((ConsumerEventService)new ConsumerEventServiceImpl(this.eventErrorService), this.transferDataSourceTemplate, consDbConfig, consTenantConfig);
    this.rocketmqConsumersGroupManager.startConsumer(consDbConfig.generateConsumerGroup(consTenantConfig), mqMessageHandler);
  }
  
  public void suspend(ConsDbConfig consDbConfig, ConsTenantConfig consTenantConfig) {
    this.rocketmqConsumersGroupManager.suspend(consDbConfig.generateConsumerGroup(consTenantConfig));
  }
  
  public void resume(ConsDbConfig consDbConfig, ConsTenantConfig consTenantConfig) {
    this.rocketmqConsumersGroupManager.resume(consDbConfig.generateConsumerGroup(consTenantConfig));
  }
  
  public DefaultMQPushConsumer createConsumer(String service, String tableName, ConsDbConfig consDbConfig, ConsTenantConfig consTenantConfig) {
    return this.rocketmqConsumersGroupManager.getOrCreateConsumer(consDbConfig.generateConsumerGroup(consTenantConfig), createTopicName(service, tableName), ConsTenantConfig.acquiredTag(consTenantConfig));
  }
  
  public void startAllConsumers() {
    this.asyncTaskExecutor.execute(() -> {
          List<String> groups = null;
          while (true) {
            try {
              groups = RocketmqClientUtils.getAllSubscriptionGroups(this.mqConfigProperties.getNameSrv());
              if (this.LOGGER.isDebugEnabled())
                this.LOGGER.debug(">>>>>>>>>>>>>>>>>>MQ GROUP TOTAL: [{}]<<<<<<<<<<<<<<<<<<", Integer.valueOf(groups.size())); 
            } catch (SubCommandException e) {
              this.LOGGER.error(e.getMessage(), (Throwable)e);
              try {
                Thread.sleep(20000L);
              } catch (InterruptedException interruptedException) {}
              continue;
            } 
            if (CollectionUtils.isNotEmpty(groups))
              try {
                startAllConsumerGroup(groups);
                break;
              } catch (Exception e) {
                if (this.LOGGER.isDebugEnabled())
                  this.LOGGER.debug("============some unknow error happen=======", e); 
              }  
          } 
        });
  }
  
  public void stopConsumer(ConsDbConfig consDbConfig, ConsTenantConfig consTenantConfig) {
    this.rocketmqConsumersGroupManager.stopConsumer(consDbConfig.generateConsumerGroup(consTenantConfig));
  }
  
  public boolean hasBeenStarted(ConsDbConfig consDbConfig, ConsTenantConfig consTenantConfig) {
    return this.rocketmqConsumersGroupManager.hasBeenStart(consDbConfig.generateConsumerGroup(consTenantConfig));
  }
  
  public static String createTopicName(String service, String tableName) {
    return service + "%" + tableName;
  }
  
  private void startAllConsumerGroup(List<String> groups) {
    for (String group : groups) {
      ConsDbConfig consDbConfig = ConsDbConfig.parseConsumerGroup(group);
      if (consDbConfig != null) {
        ConsDbConfig dbConfig = (ConsDbConfig)this.consDbConfigRepository.selectOne(consDbConfig);
        if (dbConfig != null) {
          dbConfig.setTenantId(consDbConfig.getTenantId());
          if (dbConfig.getConsDbConfigId() != null) {
            ProducerConfig producerConfig = (ProducerConfig)this.producerConfigRepository.selectByPrimaryKey(dbConfig.getProducerConfigId());
            if (this.LOGGER.isDebugEnabled())
              this.LOGGER.debug(">>>>>>>>>>>>>>>>START GROUP [{}]", group); 
            if (producerConfig != null) {
              ConsTenantConfig consTenantConfig = new ConsTenantConfig();
              consTenantConfig.setTenantId(dbConfig.getTenantId());
              startListenerWithOutUpdateOffset(producerConfig.getServiceCode(), producerConfig.getTableName(), dbConfig, consTenantConfig);
            } 
          } 
        } 
      } 
    } 
  }
  
  public void restartConsumerWithUpdateOffset(ProducerConfig producerConfig, ConsDbConfig consDbConfig, ConsTenantConfig consTenantConfig) {
    stopConsumer(consDbConfig, consTenantConfig);
    startListener(producerConfig
        .getServiceCode(), producerConfig
        .getTableName(), consDbConfig, consTenantConfig);
  }
  
  public void deleteConsumer(ProducerConfig producerConfig) {
    List<ConsDbConfig> consDbConfigList = this.transferRedisService.queryAllConsDbConfig(producerConfig);
    if (!CollectionUtils.isEmpty(consDbConfigList)) {
      consDbConfigList.forEach((dbConfig) -> {
        if (!BaseConstants.Flag.YES.equals(producerConfig.getTenantFlag())) {
          this.stopConsumer(dbConfig, (ConsTenantConfig)null);

          try {
            RocketmqClientUtils.deleteGroup(this.mqConfigProperties.getNameSrv(), dbConfig.generateConsumerGroup((ConsTenantConfig)null));
          } catch (SubCommandException var4) {
            if (this.LOGGER.isDebugEnabled()) {
              this.LOGGER.debug("delete consumer group error:", var4);
            }
          }
        } else {
          List<ConsTenantConfig> consTenantConfigList = dbConfig.getConsTenantConfigList();
          if (CollectionUtils.isEmpty(consTenantConfigList)) {
            return;
          }

          consTenantConfigList.forEach((tenantConfig) -> {
            this.stopConsumer(dbConfig, tenantConfig);

            try {
              RocketmqClientUtils.deleteGroup(this.mqConfigProperties.getNameSrv(), dbConfig.generateConsumerGroup(tenantConfig));
            } catch (SubCommandException var4) {
              if (this.LOGGER.isDebugEnabled()) {
                this.LOGGER.debug("delete consumer group error:", var4);
              }
            }

          });
        }

      });
    }
  }
}
